package com.atguigu.PracticalProject.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.PracticalProject.app.BaseAppV1;
import com.atguigu.PracticalProject.bean.PageSessionBean;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.AtguiguUtil;
import com.atguigu.PracticalProject.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficSessionDetail extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsTrafficSessionDetail().init(
                3020,
                2,
                "DwsTrafficSessionDetail",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //找出封装到bean
        SingleOutputStreamOperator<PageSessionBean> beanStream = StreamToPojo(stream);
        //开窗聚合各会话总页面浏览和总时长
        SingleOutputStreamOperator<PageSessionBean> pageSsBean = windowAndAgg(beanStream);
        //计算会话数
        SingleOutputStreamOperator<PageSessionBean> resultBean = sessionCtBean(pageSsBean);
        //写出doris
        writeToDoris(resultBean);

    }

    private void writeToDoris(SingleOutputStreamOperator<PageSessionBean> resultBean) {
        resultBean
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(bean, config);
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_traffic_session_detail"));
    }

    private SingleOutputStreamOperator<PageSessionBean> sessionCtBean(SingleOutputStreamOperator<PageSessionBean> pageSsBean) {

        return pageSsBean
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<PageSessionBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<PageSessionBean>() {
                            @Override
                            public long extractTimestamp(PageSessionBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
                )
                .keyBy(PageSessionBean::getScId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<PageSessionBean>() {
                    @Override
                    public PageSessionBean reduce(PageSessionBean bean1,
                                                  PageSessionBean bean2) throws Exception {
                        bean1.setSessionCt(bean1.getSessionCt() + bean2.getSessionCt());
                        return bean1;
                    }
                });


    }

    private SingleOutputStreamOperator<PageSessionBean> windowAndAgg(SingleOutputStreamOperator<PageSessionBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<PageSessionBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<PageSessionBean>() {
                                    @Override
                                    public long extractTimestamp(PageSessionBean pageSessionBean, long l) {
                                        return pageSessionBean.getTs();
                                    }
                                })
                )
                .keyBy(PageSessionBean::getSid)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<PageSessionBean>() {
                            @Override
                            public PageSessionBean reduce(PageSessionBean bean1,
                                                          PageSessionBean bean2) throws Exception {

                                //聚合这次回话的总浏览时长
                                bean1.setDuringTimeCt(bean1.getDuringTimeCt() + bean2.getDuringTimeCt());
                                //聚合这次回话的总浏览页面
                                bean1.setSsPageCt(bean1.getSsPageCt() + bean2.getSsPageCt());

                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<PageSessionBean, PageSessionBean, String, TimeWindow>() {
                            @Override
                            public void process(String sid,
                                                Context context,
                                                Iterable<PageSessionBean> iterable,
                                                Collector<PageSessionBean> collector) throws Exception {

                                PageSessionBean bean = iterable.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                collector.collect(bean);

                            }
                        }
                );
    }

    private SingleOutputStreamOperator<PageSessionBean> StreamToPojo(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getJSONObject("common").getString("sid"))
                .process(new KeyedProcessFunction<String, JSONObject, PageSessionBean>() {
                    @Override
                    public void processElement(JSONObject obj,
                                               Context context,
                                               Collector<PageSessionBean> collector) throws Exception {
                        //取出来源id
                        String scId = obj.getJSONObject("common").getString("sc");
                        //取出各页面浏览时长
                        Long duringTime = obj.getJSONObject("page").getLong("during_time");
                        //取出会话id
                        String sid = obj.getJSONObject("common").getString("sid");
                        Long ts = obj.getLong("ts");

                        Long ssCt = 0L;
                        //判断 一次会话的开始
                        if (!obj.containsKey(obj.getJSONObject("page").getString("last_page_id"))
                                || "mine".equals(obj.getJSONObject("page").getString("page_id"))) {
                            ssCt = 1L;

                            collector.collect(new PageSessionBean(
                                    "", "",
                                    scId, sid, "",
                                    duringTime, ssCt,
                                    1L,
                                    ts
                            ));
                        }

                        collector.collect(new PageSessionBean(
                                "", "",
                                scId, sid, "",
                                duringTime, 0L,
                                1L,
                                ts
                        ));

                    }
                });
    }
}